package com.intelli.eve.core.impl.message_bus;

import com.intelli.eve.core.api.common.Priority;
import com.intelli.eve.core.api.message.Message;
import com.intelli.ray.core.ManagedComponent;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Author: Sergey42
 * Date: 26.04.14 10:55
 */
@ManagedComponent(name = "EventBus")
public class MessageBus {

    private Map<Priority, Queue<Message>> queuesByPriority = new HashMap<>();
    private Map<Priority, AtomicInteger> counter = new HashMap<>();

    {
        for (Priority priority : Priority.values()) {
            queuesByPriority.put(priority, new ConcurrentLinkedQueue<Message>());
            counter.put(priority, new AtomicInteger(0));
        }
    }

    public void add(@Nonnull Message message) {
        System.out.println("[Bus] Adding message " + message + " at " + System.currentTimeMillis());
        Priority priority = message.getHeader().getPriority();
        queuesByPriority.get(priority).add(message);
        counter.get(priority).incrementAndGet();
    }

    @Nullable
    public Message poll() {
        Priority[] eventPriorities = Priority.values();
        for (int i = eventPriorities.length - 1; i >= 0; i--) {
            Message message = queuesByPriority.get(eventPriorities[i]).poll();
            if (message != null) {
                System.out.println("[Bus] Extracting message " + message + " at " + System.currentTimeMillis());
                counter.get(message.getHeader().getPriority()).decrementAndGet();
                return message;
            }
        }
        return null;
    }

    @Nullable
    public Message poll(Priority priority) {
        Message message = queuesByPriority.get(priority).poll();
        if (message != null) {
            System.out.println("[Bus] Extracting message " + message + " at " + System.currentTimeMillis());
            counter.get(priority).decrementAndGet();
        }
        return message;
    }

    public String getSizeInfo() {
        return counter.get(Priority.CRITICAL).intValue() + "|" +
                counter.get(Priority.HIGH).intValue() + "|" +
                counter.get(Priority.MEDIUM) + "|" +
                counter.get(Priority.LOW).intValue();
    }
}
